KAFKA-19960: Close pending tasks on shutdown. #21365

Nikita-Shupletsov · 2026-01-27T19:52:06Z

This PR fixes a bug when KS doesn't close stores if the shutdown was
triggered during rebalance where an active tasks gets converted to a
standby one and put into pendingTasksToInit

Added logic to close pending tasks to init.
Made standby task closure similar to the one for active tasks.
Added a separate method for getting standby tasks from task registry.
Added an integration test that reproduces the issue.

Reviewers: Matthias J. Sax matthias@confluent.io

Added logic to close pending tasks to init. Made standby task closure similar to the one for active tasks. Added a separate method for getting standby tasks from task registry. Added an integration test that reproduces the issue.

mjsax

Made a first pass.

Can you update the PR description adding context on what the bug exactly is, and when we hit it? It seems to be related to not closing "pending tasks", but might be good to give some more context.

.../src/test/java/org/apache/kafka/streams/integration/RebalanceTaskClosureIntegrationTest.java

...ntegration-tests/src/test/java/org/apache/kafka/streams/integration/KafkaStreamsWrapper.java

streams/src/main/java/org/apache/kafka/streams/processor/internals/Tasks.java

streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java

streams/src/main/java/org/apache/kafka/streams/processor/internals/Tasks.java

…ms/integration/RebalanceTaskClosureIntegrationTest.java Co-authored-by: Matthias J. Sax <mjsax@apache.org>

Minor refactoring.

mjsax

Some more follow ups -- also need to get input from @lucasbru...

mjsax · 2026-02-03T03:35:11Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java

+            } else {
+                standbyTasks.add(pendingTask);
+            }
+        }


Thanks for updating the PR description. It say "shutdown during rebalance when active task become standby tasks" but seems it goes either way, and the PR is actually fixing both direction?

streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java

streams/src/main/java/org/apache/kafka/streams/processor/internals/Tasks.java

lucasbru

Good find, thanks for identifying the problem.

An alternative way to fix it would be, I think, to consider pendingTasksToInit as a subset of Tasks.allTasks. Then conceptually, it may be simpler to say "close all tasks". But we'd have to check carefully what other places Tasks.allTasks and related methods are used, and if we are changing one of those places. In that case, we'd still need the "remove" fix you did.

streams/src/main/java/org/apache/kafka/streams/processor/internals/Tasks.java

lucasbru · 2026-02-03T14:15:39Z

An alternative way to fix it would be, I think, to consider pendingTasksToInit as a subset of Tasks.allTasks. Then conceptually, it may be simpler to say "close all tasks". But we'd have to check carefully what other places Tasks.allTasks and related methods are used, and if we are changing one of those places. In that case, we'd still need the "remove" fix you did.

On second thought, that would probably create more problems than it solves. But I wonder if we should rename "allTasks" to "allInitializedTasks" then?

Nikita-Shupletsov · 2026-02-03T17:22:17Z

Using drainPendingTasksToInit sounds like a good idea to me. Not sure if there are any complications though, but it would seem cleaner code wise.

the problem with drain is that we always try to remove the task we are closing from the task registry: https://github.com/Nikita-Shupletsov/kafka/blob/8ee99ee82df2cf89fbb769d26c66395fd3a63761/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java#L1405

if we drain the pending tasks, there will be no way for us to confirm it ever existed there. so we either need to relax the check in the remove method or have a separate branch for closing pending to init tasks

mjsax · 2026-02-03T20:21:47Z

maybe we would need to add a new member to TasksRegistry taking "pending tasks" which got closed?

Yes, that is what I mean by this -- after we drain the pending-tasks-to-init, we would add them into some new "pending-task-to-close" member inside TaskRegistry and update removeTask so check ownership based on this new member variable.

It's to some extend "cosmetics" as we only move all "pending task" from one collection into a different one, but it preserves the invariant we have, and makes the code cleaner.

Nikita-Shupletsov · 2026-02-03T21:14:31Z

But I wonder if we should rename "allTasks" to "allInitializedTasks" then?
@lucasbru would you mind if I did that in a followup PR?

lucasbru · 2026-02-04T10:09:25Z

But I wonder if we should rename "allTasks" to "allInitializedTasks" then?
@lucasbru would you mind if I did that in a followup PR?

No problem, it's a minor thing.

mjsax

Few follow up comments. -- Still working through the integration test, but might be good to share what I have right away.

streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java

streams/src/main/java/org/apache/kafka/streams/processor/internals/Tasks.java

streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java

streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java

streams/src/test/java/org/apache/kafka/streams/processor/internals/TasksTest.java

.../src/test/java/org/apache/kafka/streams/integration/RebalanceTaskClosureIntegrationTest.java

Co-authored-by: Matthias J. Sax <mjsax@apache.org>

Modified the test so it actually reproduces the issue. Small refactoring.

.../src/test/java/org/apache/kafka/streams/integration/RebalanceTaskClosureIntegrationTest.java

Co-authored-by: Matthias J. Sax <mjsax@apache.org>

…ms/integration/RebalanceTaskClosureIntegrationTest.java Co-authored-by: Matthias J. Sax <mjsax@apache.org>

mjsax · 2026-02-06T23:24:24Z

Thanks for the fix. -- Merged to trunk.

Cherry-picking to older branches is not clean. Can you maybe look into it and open new PRs all the way back to 4.2, 4.1, and 4.0 branches?

This PR fixes a bug when KS doesn't close stores if the shutdown was triggered during rebalance where an active tasks gets converted to a standby one and put into pendingTasksToInit * Added logic to close pending tasks to init. * Made standby task closure similar to the one for active tasks. * Added a separate method for getting standby tasks from task registry. * Added an integration test that reproduces the issue. Reviewers: Matthias J. Sax <matthias@confluent.io> --------- Co-authored-by: Matthias J. Sax <mjsax@apache.org>

This PR fixes a bug when KS doesn't close stores if the shutdown was triggered during rebalance where an active tasks gets converted to a standby one and put into pendingTasksToInit * Added logic to close pending tasks to init. * Made standby task closure similar to the one for active tasks. * Added a separate method for getting standby tasks from task registry. * Added an integration test that reproduces the issue. Reviewers: Matthias J. Sax <matthias@confluent.io> --------- Co-authored-by: Matthias J. Sax <mjsax@apache.org> Conflicts: streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java

Nikita-Shupletsov · 2026-02-07T00:22:37Z

4.0: #21421
4.1: #21422
4.2: #21423

This PR fixes a bug when KS doesn't close stores if the shutdown was triggered during rebalance where an active tasks gets converted to a standby one and put into pendingTasksToInit * Added logic to close pending tasks to init. * Made standby task closure similar to the one for active tasks. * Added a separate method for getting standby tasks from task registry. * Added an integration test that reproduces the issue. Reviewers: Matthias J. Sax <matthias@confluent.io> --------- Co-authored-by: Matthias J. Sax <mjsax@apache.org> Conflicts: streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java

This PR fixes a bug when KS doesn't close stores if the shutdown was triggered during rebalance where an active tasks gets converted to a standby one and put into pendingTasksToInit * Added logic to close pending tasks to init. * Made standby task closure similar to the one for active tasks. * Added a separate method for getting standby tasks from task registry. * Added an integration test that reproduces the issue. Reviewers: Matthias J. Sax <matthias@confluent.io> Conflicts: streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java

Nikita-Shupletsov · 2026-02-10T20:42:06Z

@lucasbru, @mjsax the follow-up PR to rename the methods in TaskRegistry: #21448
thanks in advance!

KAFKA-19960: Close pending tasks on shutdown.

ecbe338

Added logic to close pending tasks to init. Made standby task closure similar to the one for active tasks. Added a separate method for getting standby tasks from task registry. Added an integration test that reproduces the issue.

github-actions bot added triage PRs from the community streams labels Jan 27, 2026

mjsax added ci-approved and removed triage PRs from the community labels Jan 29, 2026

mjsax reviewed Jan 30, 2026

View reviewed changes

Nikita-Shupletsov and others added 2 commits January 29, 2026 21:52

Update streams/integration-tests/src/test/java/org/apache/kafka/strea…

3527671

…ms/integration/RebalanceTaskClosureIntegrationTest.java Co-authored-by: Matthias J. Sax <mjsax@apache.org>

Added a comment on what the test actually tests.

8ee99ee

Minor refactoring.

mjsax reviewed Feb 3, 2026

View reviewed changes

lucasbru reviewed Feb 3, 2026

View reviewed changes

streams/src/main/java/org/apache/kafka/streams/processor/internals/Tasks.java Outdated Show resolved Hide resolved

streams/src/main/java/org/apache/kafka/streams/processor/internals/Tasks.java Show resolved Hide resolved

Nikita-Shupletsov added 2 commits February 3, 2026 13:12

Merge branch 'apache:trunk' into KAFKA-19960

4f2bdf4

Added PendingTasksToClose.

8793347

Updated the comment to better explain the sequence of actions.

1d7586a

mjsax reviewed Feb 4, 2026

View reviewed changes

Nikita-Shupletsov and others added 4 commits February 5, 2026 14:48

Apply suggestion from @mjsax

be6646a

Co-authored-by: Matthias J. Sax <mjsax@apache.org>

Apply suggestion from @mjsax

d4f598c

Co-authored-by: Matthias J. Sax <mjsax@apache.org>

Apply suggestion from @mjsax

6b67cc2

Co-authored-by: Matthias J. Sax <mjsax@apache.org>

Added missing finals.

db98faa

Modified the test so it actually reproduces the issue. Small refactoring.

mjsax reviewed Feb 6, 2026

View reviewed changes

Nikita-Shupletsov and others added 6 commits February 5, 2026 20:37

Apply suggestion from @mjsax

f522db8

Co-authored-by: Matthias J. Sax <mjsax@apache.org>

Apply suggestion from @mjsax

df08c00

Co-authored-by: Matthias J. Sax <mjsax@apache.org>

Update streams/integration-tests/src/test/java/org/apache/kafka/strea…

8a1b56d

…ms/integration/RebalanceTaskClosureIntegrationTest.java Co-authored-by: Matthias J. Sax <mjsax@apache.org>

Update streams/integration-tests/src/test/java/org/apache/kafka/strea…

ff75096

…ms/integration/RebalanceTaskClosureIntegrationTest.java Co-authored-by: Matthias J. Sax <mjsax@apache.org>

Update streams/integration-tests/src/test/java/org/apache/kafka/strea…

ef20d15

…ms/integration/RebalanceTaskClosureIntegrationTest.java Co-authored-by: Matthias J. Sax <mjsax@apache.org>

Update streams/integration-tests/src/test/java/org/apache/kafka/strea…

6e8e87e

…ms/integration/RebalanceTaskClosureIntegrationTest.java Co-authored-by: Matthias J. Sax <mjsax@apache.org>

Update streams/integration-tests/src/test/java/org/apache/kafka/strea…

ca632f2

…ms/integration/RebalanceTaskClosureIntegrationTest.java Co-authored-by: Matthias J. Sax <mjsax@apache.org>

mjsax approved these changes Feb 6, 2026

View reviewed changes

mjsax merged commit 127a644 into apache:trunk Feb 6, 2026
23 checks passed

This was referenced Feb 7, 2026

KAFKA-19960: Close pending tasks on shutdown. (#21365) #21421

Merged

KAFKA-19960: Close pending tasks on shutdown. (#21365) #21422

Merged

KAFKA-19960: Close pending tasks on shutdown. (#21365) #21423

Merged

KAFKA-19960: Close pending tasks on shutdown. #21365

KAFKA-19960: Close pending tasks on shutdown. #21365

Conversation

Nikita-Shupletsov commented Jan 27, 2026 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

mjsax Feb 3, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

lucasbru left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

lucasbru commented Feb 3, 2026

Uh oh!

Nikita-Shupletsov commented Feb 3, 2026

Uh oh!

mjsax commented Feb 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Nikita-Shupletsov commented Feb 3, 2026

Uh oh!

lucasbru commented Feb 4, 2026

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

mjsax commented Feb 6, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Nikita-Shupletsov commented Feb 7, 2026

Uh oh!

Nikita-Shupletsov commented Feb 10, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Nikita-Shupletsov commented Jan 27, 2026 •

edited by github-actions bot

Loading

mjsax commented Feb 3, 2026 •

edited

Loading

mjsax commented Feb 6, 2026 •

edited

Loading